/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

 /***************************************************************************
 * Name: cbus_service_endpoint.c
 *
 * Purpose: service endpoint implementation for cbus.
 *
 * Developer:
 *   wen.gu , 2022-02-15
 *
 * TODO:
 *
 ***************************************************************************/

 /******************************************************************************
 **    INCLUDES
 ******************************************************************************/

#include "cbus/bus/cbus_endpoint_ops.h"

#include <string.h>

#include "cbus/core/cbus_mutex.h"
#include "cbus/core/cbus_map.h"
#include "cbus/bus/cbus_listener.h"
#include "cbus/bus/cbus_pipe.h"
#include "cbus/core/cbus_worker.h"
#include "cbus/core/cbus_string.h"
#include "cbus/core/cbus_log.h"
/******************************************************************************
 **    MACROS
 ******************************************************************************/
#define WRITE_RETRY_COUNT_DEFAULT 100

#define ACCEPT_WAIT_TIME (1000)

#define PIPE_RECV_BUF_SIZE (1024 * 16)
#define PIPE_RECV_WAIT_TIME (1000)
/******************************************************************************
 **    VARIABLE DEFINITIONS
 ******************************************************************************/


struct _cbus_service_endpoint;
typedef struct _cbus_service_endpoint cbus_service_endpoint_t;

typedef struct _service_receiver
{
    cbus_pipe_t com_pipe;
    cbus_worker_t worker_ref;
    cbus_service_endpoint_t* srv_ref;
    GPTR on_receive_opaque; /** a user opaque pointor for on receive */
    cbus_endpoint_on_receive_func on_receive;
}pipe_receiver_t;

struct _cbus_service_endpoint
{
    GSTR url;
    GPTR on_connect_opaque; /** a user opaque pointor for on connect changed */
    cbus_endpoint_on_connect_changed_func on_connect;
    cbus_listener_t listener;   
    cbus_map_t pipe_map;
    cbus_worker_t accept_worker;
};



/******************************************************************************
 **    inner FUNCTION DEFINITIONS
 ******************************************************************************/

static void pipe_receiver_free(void* ptr)
{
    pipe_receiver_t* receiver = (pipe_receiver_t*)ptr;
    cbus_pipe_destroy(receiver->com_pipe);
    receiver->com_pipe = NULL;
    cbus_worker_join(receiver->worker_ref);
    free(receiver);
}

static void connect_status_notify(cbus_service_endpoint_t* srv, pipe_receiver_t* receiver, GBOL connected)
{
    //LOGD("[%s.%d]===>: receiver: %p, com_pipe: %p, status: %s\n", __FUNCTION__, __LINE__,receiver, receiver->com_pipe, connected ? "connected" : "disconnected");
    cbus_endpoint_id endpoint_id = (cbus_endpoint_id)receiver->com_pipe;
    if (connected == GTRUE)
    {
        if (cbus_map_set_pointer(srv->pipe_map, (GU64)receiver->com_pipe, receiver, pipe_receiver_free) == GFALSE)
        {
            LOGE("insert receiver to pipe map failed, this maybe lead to memory leak\n");
        }
    }
    else
    {
        GErrc ret = cbus_worker_detach(receiver->worker_ref);

        if (G_OK == ret)
        {
            if (cbus_map_remove(srv->pipe_map, (GU64)receiver->com_pipe) == GFALSE)
            {
                LOGE("remove receiver from pipe map failed\n");
            }
        }
        else
        {
            LOGE("detach the receiver worker of service endpoint failed(0x%08x), maybe lead to memory leak. \n", ret);
        }
    }   

    if (srv->on_connect) //todo, refine me??
    {
        srv->on_connect(srv->on_connect_opaque, endpoint_id, connected);
    }   
}

static void endpoint_receive_worker(cbus_worker_t thd_worker, void* ptr)
{
    GU08 recv_buf[PIPE_RECV_BUF_SIZE];
    pipe_receiver_t* receiver = (pipe_receiver_t*)ptr;

    while (cbus_worker_running(thd_worker))
    {
        GS32 ret = cbus_pipe_read(receiver->com_pipe, recv_buf, PIPE_RECV_BUF_SIZE, PIPE_RECV_WAIT_TIME);
        //LOGD("[%s.%d]===>: pipe read ret: %d\n", __FUNCTION__, __LINE__, ret);
        if (ret > 0)
        {
            if (receiver->on_receive)
            {
                receiver->on_receive(receiver->on_receive_opaque, receiver->com_pipe, recv_buf, ret);
            }
        }
        else if ((G_ErrInterrupted == ret) || (G_ErrTimeout == ret))
        {
            //do nothing ,just try again
        }
        else 
        {
            connect_status_notify(receiver->srv_ref, receiver, GFALSE);
            break; /** todo, refine me?? */
        }  
    }
}




static pipe_receiver_t* pipe_receiver_create(cbus_service_endpoint_t* srv, cbus_pipe_t com_pipe)
{
    pipe_receiver_t* receiver = (pipe_receiver_t*)malloc(sizeof(pipe_receiver_t));

    if (receiver)
    {
        GErrc ret = cbus_worker_create(&receiver->worker_ref, "srv ep pipe receiver", endpoint_receive_worker, receiver); 

        if (G_OK != ret)
        {
            free(receiver);
            return NULL;
        }

        ret = cbus_worker_start(receiver->worker_ref);

        if (G_OK != ret)
        {
            cbus_worker_join(receiver->worker_ref);
            free(receiver);
            return NULL;
        }

        receiver->com_pipe = com_pipe;
        receiver->srv_ref = srv;
    }

    return receiver;
}

static void endpoint_accept_worker(cbus_worker_t thd_worker, void* ptr)
{
    cbus_service_endpoint_t* srv = (cbus_service_endpoint_t*)ptr; 
    while (cbus_worker_running(thd_worker))
    {
        cbus_pipe_t com_pipe = NULL;
        GErrc ret = cbus_listener_accept(srv->listener, &com_pipe, ACCEPT_WAIT_TIME);

        if (G_OK == ret)
        {
            pipe_receiver_t* receiver = pipe_receiver_create(srv, com_pipe);

            if (receiver)
            {
                connect_status_notify(srv, receiver, GTRUE);
            }
        }

        if (!srv->listener)
        {
            break;
        }
    }
}

/******************************************************************************
 **    FUNCTION DEFINITIONS
 ******************************************************************************/
static void cbus_service_endpoint_finish(GPHD handle)
{
    if (handle)
    {
        cbus_service_endpoint_t* srv = (cbus_service_endpoint_t*)handle;      
        cbus_listener_destroy(srv->listener);
        srv->listener = NULL;
        cbus_worker_join(srv->accept_worker);

        cbus_map_destroy(srv->pipe_map);        

        if (srv->url)
        {
            free(srv->url);
        }
        
        free(handle);
    }
}

static GErrc cbus_service_endpoint_initialize(GPHD* handle, const GSTR url)
{
    if (!handle || !url)
    {
        return G_ErrBadParameter;
    }

    cbus_transport_scheme_t scheme;
    GErrc ret = cbus_transport_get_scheme_name_by_url(url, scheme);

    if (G_OK != ret)
    {
        return ret;
    }

    cbus_service_endpoint_t* srv = (cbus_service_endpoint_t*)malloc(sizeof(cbus_service_endpoint_t));

    if (!srv)
    {
        LOGE("alloc service endpoint instance failed\n");
        return G_ErrInsufficientResources;
    }

    memset(srv, 0, sizeof(cbus_service_endpoint_t));
    ret = cbus_listener_create(&srv->listener, scheme);
    //LOGD("[%s.%d]===>:listener:%p, ret:0x%08x\n", __FUNCTION__, __LINE__, srv->listener, ret);
    if (G_OK != ret)
    {
        free(srv);
        return ret;
    }
    
    ret = cbus_worker_create(&srv->accept_worker,"endport accept", endpoint_accept_worker, srv);

    if (G_OK != ret)
    {
        cbus_service_endpoint_finish(srv);
        return ret;
    }

    srv->pipe_map = cbus_map_create(CMV_TYPE_Pointer, CBUS_MAP_DEFAULT_CAPACITY);

    if (!srv->pipe_map)
    {
        cbus_service_endpoint_finish(srv);
        return G_ErrInsufficientResources;
    }

    srv->url = cbus_str_dup(url);
    *handle = (GPHD)srv;

    return G_OK;
}

static GErrc cbus_service_endpoint_start(GPHD handle)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }

    cbus_service_endpoint_t* srv = (cbus_service_endpoint_t*)handle;
    GErrc ret = cbus_worker_start(srv->accept_worker);

    if (G_OK == ret)
    {
        return cbus_listener_bind(srv->listener, srv->url);
    }
    
    return ret;
}

static GErrc cbus_service_endpoint_stop(GPHD handle)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }

    cbus_service_endpoint_t* srv = (cbus_service_endpoint_t*)handle;

    GErrc ret = cbus_worker_pause(srv->accept_worker);

    if (G_OK == ret)
    {
        cbus_map_clear(srv->pipe_map);
        return cbus_listener_unbind(srv->listener);
    }

    return ret;    
}

static GErrc cbus_service_endpoint_set_on_connect(GPHD handle, cbus_endpoint_on_connect_changed_func on_connect, GPTR opaque)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }

    cbus_service_endpoint_t* srv = (cbus_service_endpoint_t*)handle;

    srv->on_connect = on_connect;
    srv->on_connect_opaque = opaque;
    return G_OK;
}

static GErrc cbus_service_endpoint_set_on_recv(GPHD handle, cbus_endpoint_id endpoint_id, cbus_endpoint_on_receive_func on_receive, GPTR opaque)
{
    if (!handle || !on_receive)
    {
        return G_ErrBadParameter;
    }

    cbus_service_endpoint_t* srv = (cbus_service_endpoint_t*)handle;

    pipe_receiver_t* receiver;

    if (cbus_map_find_pointer(srv->pipe_map, (GU64)endpoint_id, (const void**)&receiver) == GFALSE)
    {
        return G_ErrNotFound;
    }

    receiver->on_receive        = on_receive;
    receiver->on_receive_opaque = opaque;

    return G_OK;
}

static GErrc cbus_service_endpoint_send(GPHD handle, cbus_endpoint_id endpoint_id, const GBUF data, GU32 size)
{
    if (!handle || !endpoint_id)
    {
        return G_ErrBadParameter;
    }

    //LOGD("[%s.%d]===>: ep id: 0x%lx, data: %p, size: %d\n", __FUNCTION__, __LINE__, (GU64)endpoint_id, data, size);
    cbus_service_endpoint_t* srv = (cbus_service_endpoint_t*)handle;

    /** todo, find receiver , if write failed(e.g. return G_ErrDisconnect), then remove receiver from pipe map*/
    return cbus_pipe_write((cbus_pipe_t)endpoint_id, data, size);
}



//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

static const cbus_endpoint_ops g_srv_endpoint_ops = 
{

    CBUS_EP_ROLE_SERVICE,
    CBUS_EP_VERSION,
    "default",
    cbus_service_endpoint_initialize,
    cbus_service_endpoint_finish,
    cbus_service_endpoint_start,
    cbus_service_endpoint_stop,
    cbus_service_endpoint_set_on_connect, 
    cbus_service_endpoint_set_on_recv,
    cbus_service_endpoint_send,

    
};


GErrc cbus_service_endpoint_default_ctor(void)
{
    return cbus_endpoint_ops_register(&g_srv_endpoint_ops);
}
